package com.dk.mq;

import com.dk.service.FreemarkerService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @ClassName RocketMQConsumer
 * @Deacription TODO
 * @Author hqq
 * @Date 2021/11/1 16:37
 * @Version 1.0
 **/
@Component
@RocketMQMessageListener(topic="id",consumerGroup = "myConsumer-group")
public class RocketMQConsumer implements RocketMQListener<Long>, RocketMQPushConsumerLifecycleListener {

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private FreemarkerService freemarkerService;

    @Override
    public void onMessage(Long aLong) {

    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt messageExt = list.get(0);
                int reconsumeTimes = messageExt.getReconsumeTimes();//重试次数
                try{
                    Boolean mqMessage = redisTemplate.boundHashOps("mqMessage").hasKey(messageExt.getMsgId());
                    if(!mqMessage){
                        String str = new String(messageExt.getBody());//消息内容
                        System.out.println("-----" + Long.parseLong(str));
                        freemarkerService.createPage(Long.parseLong(str));  //消费消息
                        System.out.println("消费成功");
                    }
                } catch (Exception e){
                    if(reconsumeTimes > 1){
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER; //消费成功
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER; //若消费失败,稍后再消费
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功
            }
        });
    }
}
